package com.xxl.job.admin.core.thread;

import com.xxl.job.admin.core.complete.XxlJobCompleter;
import com.xxl.job.admin.core.conf.XxlJobAdminConfig;
import com.xxl.job.admin.core.model.XxlJobLog;
import com.xxl.job.admin.core.util.I18nUtil;
import com.xxl.job.core.biz.model.HandleCallbackParam;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.util.DateUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Date;
import java.util.List;
import java.util.concurrent.*;

/**
 * job lose-monitor instance
 *
 * @author xuxueli 2015-9-1 18:05:56
 */
public class JobCompleteHelper {

  private static Logger logger = LoggerFactory.getLogger(JobCompleteHelper.class);

  private static JobCompleteHelper instance = new JobCompleteHelper();

  public static JobCompleteHelper getInstance() {
    return instance;
  }

  // ---------------------- monitor ----------------------

  private ThreadPoolExecutor callbackThreadPool = null;
  private Thread monitorThread;
  private volatile boolean toStop = false;

  public void start() {

    // for callback
    callbackThreadPool = new ThreadPoolExecutor(
        2,
        20,
        30L,
        TimeUnit.SECONDS,
        new LinkedBlockingQueue<Runnable>(3000),
        new ThreadFactory() {
          @Override
          public Thread newThread(Runnable r) {
            return new Thread(r,
                "xxl-job, admin JobLosedMonitorHelper-callbackThreadPool-" + r.hashCode());
          }
        },
        new RejectedExecutionHandler() {
          @Override
          public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            r.run();
            logger.warn(
                ">>>>>>>>>>> xxl-job, callback too fast, match threadpool rejected handler(run now).");
          }
        });

    // for monitor
    // 不执行任务超时10min执行器健康检查
    /*monitorThread = new Thread(new Runnable() {

      @Override
      public void run() {

        // wait for JobTriggerPoolHelper-init
        try {
          TimeUnit.MILLISECONDS.sleep(50);
        } catch (InterruptedException e) {
          if (!toStop) {
            logger.error(e.getMessage(), e);
          }
        }

        // monitor
        while (!toStop) {
          try {
            // 任务结果丢失处理：调度记录停留在 "运行中" 状态超过10min，且对应执行器心跳注册失败不在线，则将本地调度主动标记失败；
            Date losedTime = DateUtil.addMinutes(new Date(), -10);
            List<Long> losedJobIds = XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao()
                .findLostJobIds(losedTime);

            if (losedJobIds != null && losedJobIds.size() > 0) {
              for (Long logId : losedJobIds) {

                XxlJobLog jobLog = new XxlJobLog();
                jobLog.setId(logId);

                jobLog.setHandleTime(new Date());
                jobLog.setHandleCode(ReturnT.FAIL_CODE);
                jobLog.setHandleMsg(I18nUtil.getString("joblog_lost_fail"));

                XxlJobCompleter.updateHandleInfoAndFinish(jobLog);
              }

            }
          } catch (Exception e) {
            if (!toStop) {
              logger.error(">>>>>>>>>>> xxl-job, job fail monitor thread error:{}", e);
            }
          }

          try {
            TimeUnit.SECONDS.sleep(60);
          } catch (Exception e) {
            if (!toStop) {
              logger.error(e.getMessage(), e);
            }
          }

        }

        logger.info(">>>>>>>>>>> xxl-job, JobLosedMonitorHelper stop");

      }
    });
    monitorThread.setDaemon(true);
    monitorThread.setName("xxl-job, admin JobLosedMonitorHelper");
    monitorThread.start();*/
  }

  public void toStop() {
    toStop = true;

    // stop registryOrRemoveThreadPool
    callbackThreadPool.shutdownNow();

    // stop monitorThread (interrupt and wait)
    // 不执行任务超时10min执行器健康检查
    /*monitorThread.interrupt();
    try {
      monitorThread.join();
    } catch (InterruptedException e) {
      logger.error(e.getMessage(), e);
    }*/
  }

  // ---------------------- helper ----------------------

  public ReturnT<String> callback(List<HandleCallbackParam> callbackParamList) {

    callbackThreadPool.execute(new Runnable() {
      @Override
      public void run() {
        for (HandleCallbackParam handleCallbackParam : callbackParamList) {
          ReturnT<String> callbackResult = callback(handleCallbackParam);
          logger.debug(
              ">>>>>>>>> JobApiController.callback {}, handleCallbackParam={}, callbackResult={}",
              (callbackResult.getCode() == ReturnT.SUCCESS_CODE ? "success" : "fail"),
              handleCallbackParam, callbackResult);
        }
      }
    });

    return ReturnT.SUCCESS;
  }

  private ReturnT<String> callback(HandleCallbackParam handleCallbackParam) {
    // valid log item
    XxlJobLog log = XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao()
        .load(handleCallbackParam.getLogId());
    if (log == null) {
      return new ReturnT<String>(ReturnT.FAIL_CODE, "log item not found.");
    }
    if (log.getHandleCode() > 0) {
      return new ReturnT<String>(ReturnT.FAIL_CODE,
          "log repeate callback.");     // avoid repeat callback, trigger child job etc
    }

    // handle msg
    StringBuffer handleMsg = new StringBuffer();
    if (log.getHandleMsg() != null) {
      handleMsg.append(log.getHandleMsg()).append("<br>");
    }
    if (handleCallbackParam.getHandleMsg() != null) {
      handleMsg.append(handleCallbackParam.getHandleMsg());
    }

    // success, save log
    log.setHandleTime(new Date());
    log.setHandleCode(handleCallbackParam.getHandleCode());
    log.setHandleMsg(handleMsg.toString());
    XxlJobCompleter.updateHandleInfoAndFinish(log);

    return ReturnT.SUCCESS;
  }


}
